import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.Scanner;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

public class Map extends MapReduceBase 
	                          implements Mapper<IntWritable, BytesWritable, Text, Text> {
	 
		@Override
		public void map(IntWritable key, BytesWritable value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
	  	   Scanner read = new Scanner(new File("./query.dat"));
	  	   byte[] line= value.getBytes();
	  	   System.err.println("Map got split");
	  	   while (read.hasNext())
	  	   {
	  		   String s = read.nextLine();
	  		   byte[] target = s.getBytes();
	  		   int process=0;
		  	   if (line.length==67108864+99)
		  	   {
		  		   process = 99;
		  	   }
		  	   else process = target.length+98;
		  	   
	  		   boolean match = false;
	  		   
	  		   String word = ((FileSplit)reporter.getInputSplit()).getPath().getName()+","+s;
	  		   for (int i=0;i<line.length-process;i++)
	  		   {
	  			   byte[] compare = Arrays.copyOfRange(line, i, i+target.length);
	  			   if(Arrays.equals(compare, target))
	  			   {
	  				   Integer val = new Integer(i+key.get());
	  				   output.collect(new Text(word),new Text(val.toString()));
	  				   match = true;
	  			   }
	  			   if(!match&&i==line.length-process-1)
	  			   {
	  				   output.collect(new Text(word),new Text(""));
	  			   }   
	  		   }
	  		   System.err.println("A query completed");
	  	   }
	  	   System.err.println("Map Completed");
	  	   read.close();
		}
	  }